package com.flute.haflute.jobbox.common;

import java.rmi.RemoteException;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Vector;
import java.util.concurrent.atomic.AtomicInteger;

import com.flute.haflute.common.ClusterContext;
import com.flute.haflute.jobbox.common.tools.OSUtil;

/**
 * 分布式平台集群环境 保存在此
 * 
 */
public class CloudContext {

	/** 每个节点的子节点数，保存为<节点ID，子节点的数目> */
	private static Map<String, Integer> numDeamonsPerNode = new HashMap<String, Integer>();

	/** 所有节点名的列表 */
	private static List<String> chids = new Vector<String>();

	/** 所有退出集群节点的列表 */
	private static List<String> removedChilds = new Vector<String>();

	/** 所有新加入节点的列表 */
	private static List<String> freshChids = new Vector<String>();

	/** 保存Hadoop的一些参数，集成时需要 */
	private static Properties hadoop_props = new Properties();

	/** 平台运行的根目录 */
	private static String rootDir = "";

	/** 集群主节点IP或者Name */
	private static String cloudMaster = "localhost";

	/** 预定义的节点集 */
	private static Map<String, String> definedNodeSet;

	/** 活动节点信息，每一个节点的活动子节点数目 <节点ID，子节点的数目> 对 */
	private static Map<String, AtomicInteger> activeNodesInfo = new Hashtable<String, AtomicInteger>();

	/** 当前的子节点全局号码池 */
	private static int numChildDaemons = 0;

	/** 子节点唯一标识的格式，不能轻易改动 */
	private static String childIdPatternTag = "-";

	/** 子节点唯一标识的格式，不能轻易改动 */
	private static String childIdPattern = "ip" + childIdPatternTag + "dno";

	/** RMI监听端口 */
	private static int rmiPort = 1099;

	/** 心跳检测间隔，默认为一分钟 */
	private static int heartbeatCheckInterval = 60000;

	/** 心跳检测失败的次数，达到该次数后，主节点将从集群中删除子节点 */
	private static int heartbeatFailedTimes = 3;

	/** 心跳检测重试的次数 */
	private static int heartbeatTriedTimes = 3;

	/** 心跳侦听的端口 */
	private static int heartbeatListenPort = 2000;

	/**
	 * 得到每个节点能够运行任务的后台子节点（进程）数
	 */
	public static int getNumDeamons4Node(String nodeIp) {
		Integer numDaemons = numDeamonsPerNode.get(nodeIp);
		if (numDaemons == null) {
			// origin CloudConstants.DEFAULT_NUM_DAEMONS;
			numDaemons = 400;
			setNumDaemons4Node(nodeIp, numDaemons);
		}
		return numDaemons;
	}

	/** 得到节点列表 */
	public static List<String> getChids() {
		return chids;
	}

	/** 得到节点列表 */
	public static List<String> getRemovedChids() {
		return removedChilds;
	}

	/** 得到节点列表 */
	public static List<String> getFreshChids() {
		return freshChids;
	}

	/**
	 * 增加一个节点，如果已经存在则返回，否则增加该节点已经连接的子节点的数量
	 */
	public static synchronized void addChild(String child) {
		if (chids.contains(child))
			return;
		chids.add(child);
		if (removedChilds.contains(child)) {
			removedChilds.remove(child);
		} else {
			freshChids.add(child);
		}

		String ip = getIpFromChildName(child);
		AtomicInteger numOfDaemon = activeNodesInfo.get(ip);
		if (numOfDaemon == null) {
			numOfDaemon = new AtomicInteger(0);
			activeNodesInfo.put(ip, numOfDaemon);
		}
		numOfDaemon.incrementAndGet();
	}

	/**
	 * 删除一个节点
	 */
	public static synchronized void delChild(String child) {
		if (chids.contains(child)) {
			chids.remove(child);
			removedChilds.add(child);
		}

		String ip = getIpFromChildName(child);
		AtomicInteger numOfDaemon = activeNodesInfo.get(ip);
		if (numOfDaemon == null)
			return;
		numOfDaemon.decrementAndGet();
		if (numOfDaemon.intValue() <= 0)
			activeNodesInfo.remove(ip);
	}

	/**
	 * 从子节点ID中获取IP地址
	 */
	public static String getIpFromChildName(String child) {
		return child.substring(0, child.indexOf(childIdPatternTag));
	}

	/**
	 * 获取集群节点数
	 */
	public static int getNumOfChilds() {
		return chids.size();
	}

	/**
	 * 获取分布式平台的根目录
	 */
	public static String getRootDir() {
		if ("".equals(rootDir)) {
			/*
			 * rootDir = System.getProperty("user.dir"); rootDir = (new
			 * StringBuilder(rootDir)).append(File.separator).toString();
			 */
			rootDir = ClusterContext.class.getResource("/").getPath()
					.toString();
		}
		return rootDir;
	}

	/**
	 * 设置分布式平台的根目录
	 */
	public static void setRootDir(String dir) {
		rootDir = dir;
	}

	public synchronized static Properties getHadoopProperties() {
		if (hadoop_props == null)
			hadoop_props = new Properties();
		return hadoop_props;
	}

	public static void setHadoopProperties(Properties props) {
		hadoop_props = props;
	}

	/** 设置集群主节点地址 */
	public static void setCloudMaster(String master) {
		cloudMaster = master;
	}

	/** 取得主节点的地址 */
	public static String getCloudMaster() {
		return cloudMaster;
	}

	/** 取得预定义的节点集 */
	public static Map<String, String> getDefinedNodeSet() {
		return definedNodeSet;
	}

	/** 预定义的节点集 */
	public static void setDefinedNodeSet(Map<String, String> cloudChildren) {
		CloudContext.definedNodeSet = cloudChildren;
	}

	/** 得到主节点的ID，默认为IP地址 */
	public static String getMyID() {
		return OSUtil.getLocalIP(true);
	}

	/** 得到当前节点的IP地址 */
	public static String getLocalIP() {
		return OSUtil.getLocalIP(true);
	}

	/** 增加一个预定义节点 */
	public static void addDefinedNode(String name, String ip) {
		synchronized (definedNodeSet) {
			definedNodeSet.put(name, ip);
		}
	}

	/** 取得主节点RMI监听接口 */
	public static int getRMIPort() {
		return rmiPort;
	}

	/** 设置主节点RMI监听接口 */
	public static void setRmiPort(int rmiPort) {
		CloudContext.rmiPort = rmiPort;
	}

	/**
	 * 取得子节点ID格式
	 * 
	 * @see childIdPattern
	 */
	public static String getChildIdPattern() {
		return childIdPattern;
	}

	/**
	 * 取得子节点的全局号
	 */
	public static int getChildDaemonNo(String ip) throws RemoteException {
		AtomicInteger activeDaemons4Node = activeNodesInfo.get(ip);
		if (activeDaemons4Node == null) {
			activeDaemons4Node = new AtomicInteger(0);
			activeNodesInfo.put(ip, activeDaemons4Node);
		} else if (activeDaemons4Node.intValue() >= CloudContext
				.getNumDeamons4Node(ip)) {
			throw new RemoteException("too many Daemons for this node:" + ip);
		}
		numChildDaemons++;
		return numChildDaemons;
	}

	/** 取得心跳消息发送的时间间隔 */
	public static int getHBInterval() {
		return heartbeatCheckInterval;
	}

	/** 取得心跳消息允许的失败次数 */
	public static int getHBFailedTimes() {
		return heartbeatFailedTimes;
	}

	/** 取得心跳不正常时，每次尝试重新连接的次数 */
	public static int getMaxHelloTimes4Child() {
		return heartbeatTriedTimes;
	}

	/** 取得心跳检测的间隔 */
	public static void setHBCheckInterval(int checkInterval) {
		heartbeatCheckInterval = checkInterval * 1000;
	}

	/** 设置心跳失败的周期数目 */
	public static void setHBFailedTimes(int failedTimes) {
		heartbeatFailedTimes = failedTimes;
	}

	/** 设置心跳不正常时，每次尝试重新连接的次数 */
	public static void setHBMaxHelloTimes4Child(int triedTimes) {
		heartbeatTriedTimes = triedTimes;
	}

	/** 设置心跳侦听端口 */
	public static void setHBListenPort(int port) {
		heartbeatListenPort = port;
	}

	/** 取得主节点心跳侦听端口 */
	public static int getHBListenPort() {
		return heartbeatListenPort;
	}

	/** 设置每个节点的子节点进程数目 */
	public static void setNumDaemons4Node(String nodeIp, int numDaemons) {
		numDeamonsPerNode.put(nodeIp, numDaemons);
	}
}
